package com.freez.spark.core.demo;

import com.clearspring.analytics.util.Lists;
import com.freez.spark.tool.TupleUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.*;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: AnalysisShop.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:17
 * Description: 好友推荐，共同好友
 */
public class Friends {
    public static void main(String[] args) {
        String appName1 = "SparkDemo-commonFriend";
        String master = "local";
        String path = "./src/main/resources/data/core/demo/friends.txt";
        commonFriend(appName1, master, path);
        String appName2 = "SparkDemo-RecommendFriend";
        recommendFriend(appName2, master, path);
    }

    /**
     * 查找共同好友
     *
     * @param appName Application的名字
     * @param master  设置Spark程序的运行模式
     * @param path    源文件路径
     */
    public static void commonFriend(String appName, String master, String path) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> txtRDD = jsc.textFile(path);
        JavaPairRDD<Tuple2<String, String>, Iterable<Iterable<String>>> groupByKey = txtRDD.flatMapToPair(s -> {
            String[] split = s.split(":");
            String person = split[0];
            String[] friendSet = split[1].split(",");
            //处理好友集合只有一个时侯的情况
            if (friendSet.length == 1) {
                Tuple2<String, String> key = buildSortPair(person, friendSet[0]);
                Tuple2<Tuple2<String, String>, Iterable<String>> tempResult = TupleUtils.T2(key, new ArrayList<>());
                return Arrays.asList(tempResult).iterator();
            }
            List<String> friendList = Arrays.asList(friendSet);
            List<Tuple2<Tuple2<String, String>, Iterable<String>>> resultList = new ArrayList<>();
            for (String friend : friendList) {
                Tuple2<String, String> key = buildSortPair(person, friend);
                Tuple2<Tuple2<String, String>, Iterable<String>> tempTuple = TupleUtils.T2(key, friendList);
                resultList.add(tempTuple);
            }
            return resultList.iterator();
        }).groupByKey();
        //对groupByKey后
        JavaPairRDD<Tuple2<String, String>, Iterable<String>> mapValues = groupByKey.mapValues(v1 -> {
            //ArrayList<Iterable<String>> iterables = new ArrayList<Iterable<String>>(v1);
            // 集合求交集
            List<Iterable<String>> list = Lists.newArrayList(v1);
            List<String> aList = Lists.newArrayList(list.get(0));
            List<String> bList = Lists.newArrayList(list.get(1));
            aList.retainAll(bList);
            return aList;
        });
        mapValues.foreach(s -> System.out.println(s));
//        mapValues.saveAsTextFile("hdfs://masterNode1:9000/demo/result/");
        jsc.close();
    }

    /**
     * 好友推荐
     *
     * @param appName Application的名字
     * @param master  设置Spark程序的运行模式
     * @param path    源文件路径
     */
    public static void recommendFriend(String appName, String master, String path) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> txtRDD = jsc.textFile(path);
        JavaPairRDD<String, Tuple2<String, String>> pairRDD = txtRDD.flatMapToPair(s -> {
            String[] split = s.split(":");
            String person = split[0];
            String[] friendSet = split[1].split(",");
            List<Tuple2<String, Tuple2<String, String>>> mapOutList = new ArrayList<>();
            List<String> friendsList = new ArrayList<>();
            //直接好友
            for (String friend : friendSet) {
                String toUser = friend;
                friendsList.add(toUser);
                Tuple2<String, String> directFT = TupleUtils.T2(toUser, "XX");
                mapOutList.add(TupleUtils.T2(person, directFT));
            }
            for (int i = 0; i < friendsList.size(); i++) {
                for (int j = i + 1; j < friendsList.size(); j++) {
                    //间接好友1
                    Tuple2<String, String> possibleFriend1 = TupleUtils.T2(friendsList.get(j), person);
                    mapOutList.add(TupleUtils.T2(friendsList.get(i), possibleFriend1));
                    //间接好友2
                    Tuple2<String, String> possibleFriend2 = TupleUtils.T2(friendsList.get(i), person);
                    mapOutList.add(TupleUtils.T2(friendsList.get(j), possibleFriend2));
                }
            }
            return mapOutList.iterator();
        });
        JavaPairRDD<String, Iterable<Tuple2<String, String>>> groupByKey = pairRDD.groupByKey();
        JavaPairRDD<String, String> resultRDD = groupByKey.mapValues(value -> {
            Map<String, List<String>> commonFMap = new HashMap<>();
            for (Tuple2<String, String> item : value) {
                String toUser = item._1;
                String commonFriend = item._2;
                boolean isExits = "XX".equals(commonFriend);
                if (commonFMap.containsKey(toUser)) {
                    if (isExits) {
                        commonFMap.put(commonFriend, null);
                    } else if (commonFMap.get(toUser) != null) {
                        commonFMap.get(toUser).add(commonFriend);
                    }
                } else {
                    if (isExits) {
                        commonFMap.put(commonFriend, null);
                    } else {
                        List<String> list = new ArrayList<>(Arrays.asList(commonFriend));
                        commonFMap.put(toUser, list);
                    }
                }
            }
            return buildResult(commonFMap);
        });
        resultRDD.foreach(s -> System.out.println(s));
        jsc.close();
    }

    /**
     * 处理 A-B 和 B-A 的情形
     *
     * @param a
     * @param b
     * @return Tuple2<String, String> 即人名key
     */
    public static Tuple2<String, String> buildSortPair(String a, String b) {
        return a.compareTo(b) > 0 ? TupleUtils.T2(b, a) : TupleUtils.T2(a, b);
    }

    public static String buildResult(Map<String, List<String>> friends) {
        StringBuffer sb = new StringBuffer();
        for (Map.Entry<String, List<String>> item : friends.entrySet()) {
            if (item.getValue() == null) { //直接好友
                continue;
            }
//            sb.append("和 "+item.getKey()).append(" 的共同好友(").append(item.getValue().size()).append("个:").append(item.getValue()).append("),");
            sb.append(item.getKey()).append(item.getValue()).append(",");
        }
        return sb.toString();
    }

}
